﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Text;
using System.Text.Json.Serialization;
using System.Threading.Tasks;

namespace Zhaoxi.KafkaServer
{
	public class a
	{
		//腾讯都在招.net 我们之前的公司，一千多人的，都是.net
	}
	class ConfulentKafka
	{

		/// <summary>
		/// 发送事件
		/// </summary>
		/// <param name="event"></param>
		public static async Task Produce(string brokerlist, string topicname, string content)
		{


			string brokerList = brokerlist;
			string topicName = topicname;

			var config = new ProducerConfig
			{

				BootstrapServers = brokerList,
				//幂等性 针对于我们的一个分区
				//	EnableIdempotence = true,
				Acks = Acks.All,
				//LingerMs = 10000,
				BatchNumMessages = 1,//控制条数，当条数到达这个限额的时候，一次性发送，还有时间控制，如果当前的条数不够，但是等待的时间够了，则也可以打包发送
				MessageSendMaxRetries = 3,//如果发送失败，则重试三次
										  //  Partitioner = Partitioner.Random


			};

			using (var producer = new ProducerBuilder<string, string>(config)
				//.SetValueSerializer(new CustomStringSerializer<string>())
				//	.SetStatisticsHandler((o, json) =>
				//{
				//	Console.WriteLine("json");
				//	Console.WriteLine(json);
				//})
				.Build())
			{

				Console.WriteLine("\n-----------------------------------------------------------------------");
				Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");
				Console.WriteLine("-----------------------------------------------------------------------");
				try
				{
					//消息的发送端//目前是通过key 分配数据放到那个分区
					// 如果所有的key都是一致，则所有的数据都放到同一分区里面去了

					var deliveryReport = await producer.ProduceAsync(
					topicName, new Message<string, string> { Key = "", Value = content });
					Console.WriteLine($"delivered to: {deliveryReport.TopicPartitionOffset}");
				}
				catch (ProduceException<string, string> e)
				{
					Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}]");
				}
			}
		}
	}
}